package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular

import java.lang.Exception

import org.apache.spark.sql.{AnalysisException, DataFrame, SparkSession}


class brand_join_res {
  /**
   * @param sourceData      :DataFrame  跑完分类的【数据】
   * @param BrandResultPath ：String  跑完品牌的数据结果路径
   * @param year            ：int                 年
   * @param month           ：int                月
   * @param platformName    ：string     平台名称
   * @param spark
   */

  def brandJoinResult(sourceData: DataFrame, BrandResultPath: String,
                      year: Int, month: Int, platformName: String, spark: SparkSession) {
    // ************************************************************************************************************
    val timeStamp = s"${year}年${month}月"
    val brandTableOldPath = s"s3a://o2o-dimension-table/brandName_table/${year}/${month}/${platformName}/"
    val brandTableResult = s"s3a://o2o-dimension-table/brandName_table/${year}/${month + 1}/${platformName}/"
    val lastbrandTableResult = s"s3a://o2o-dimension-table/brandName_table/${year}/${month - 1}/${platformName}/"
    val brandNewAddTable = s"s3a://o2o-dimension-table/newAddBrand/${year}/${month}/${platformName}/"
    // ************************************************************************************************************

    println("--------------提取新增品牌--------------------------")
    var brandTableOld = spark.sqlContext.emptyDataFrame
    val addBrand = new utils.newAdd_brand_util
    //    读取品牌表
    try {
      brandTableOld = spark.read.json(brandTableOldPath)
      println("本月品牌读取")
    }catch{
      case ex: AnalysisException =>{
        try {
          brandTableOld = spark.read.json(lastbrandTableResult)
          println("读取本月品牌error")
          println("读取上月品牌")
        }catch{
          case ex: AnalysisException =>{
            println("Missing file exception")
            println("读取上月品牌error")
            println("新建品牌")
            brand_join_res_new.brandJoinResult(sourceData, BrandResultPath, year, month, platformName, spark)
            brandTableOld = spark.read.json(brandTableOldPath)
          }
        }
//        println("Missing file exception")
//        brand_join_res_new.brandJoinResult(sourceData, BrandResultPath, year, month, platformName, spark)
//        brandTableOld = spark.read.json(brandTableOldPath)
      }
    }
    if(brandTableOld.count()==0){
      throw new NullPointerException
    }
    val new_brand = addBrand.new_brand_excat(sourceData, brandTableOld, spark, timeStamp)
    //    统计一下新增品牌条数，如果没有新品牌直接关联品牌入库
    val brandNewCount = new_brand.dropDuplicates("brandCcId").count()
    if (brandNewCount > 0) {
      val brandNewAddToOld = new utils.BrandNewAddToOld
      brandNewAddToOld.newToOld(new_brand, brandTableOld, spark, brandTableResult, BrandResultPath, sourceData, platformName, year, month, brandNewAddTable, timeStamp)
    } else {
      brandTableOld.select("brandCcId", "brandName", "brandName_cn", "brandName_en", "brandValueId", "brand_isLaoZiHao", "brand_type", "firstCategoryId", "platform", "timeStamp", "brand_state").repartition(1).write.mode("overwrite").json(brandTableResult)
      val brand = new utils.BrandJoin
      val brandResultData = brand.brandResult(sourceData, spark, brandTableOld)
      brandResultData.coalesce(30).write.orc(BrandResultPath)
    }
  }
}


